package com.diana.config;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Component
public class TestRocketConsumer extends RocketMqConsumerConfig implements ApplicationListener<ContextRefreshedEvent> {

    private static final Logger log = LoggerFactory.getLogger(TestRocketConsumer.class);
    @Override
    public ConsumeConcurrentlyStatus dealBody(List<MessageExt> list) {
        log.info("消息数量:"+list.size());
        for (MessageExt m:list){
            String message = null;
            try {
                message = new String(m.getBody(),"UTF-8");
            } catch (UnsupportedEncodingException e) {
                log.error("字符串转换异常:{}",e);
            }
            log.info("消息内容:"+message);
            int reconsumeTimes = m.getReconsumeTimes();
            log.info(reconsumeTimes+"");
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        try {
            super.listener("dianaboot","tag1");
        }catch (MQClientException e){
            log.error("消费者监听启动失败:{}",e);
        }
    }
}
